package study.datastructure.queue.impl;/**
 * @program: datastructure
 * @author: lcy
 * @create: 2024-11-03 17:01
 */

/**
 2024/11/3,
 */

import study.datastructure.queue.BlockingQueue;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 双锁实现
 */
public class TestBlockingQueue2<E>implements BlockingQueue<E> {
    public final  E[] queue;

    //实际大小
    private AtomicInteger size=new AtomicInteger(0);
    //尾部
    int tail=0;

    //队列头
    int head=0;

    private boolean isFull() {return size.get() == queue.length;}

    private boolean isEmpty() { return size.get() == 0; }
    public TestBlockingQueue2(int capacity) {
        queue= (E[]) new Object[capacity];
    }

    ReentrantLock headLock = new ReentrantLock();
    ReentrantLock tailLock = new ReentrantLock();
    Condition tailsWait = tailLock.newCondition(); //待插入阻塞线程的集合
    Condition headsWait = headLock.newCondition(); //待取值阻塞线程的集合

//级联 处理
//
//并行 加锁
    @Override
    public void offer(E e) throws InterruptedException {
        //上尾锁插入
        tailLock.lockInterruptibly();
        int curr;
        try {
            while (isFull()) {
                tailsWait.await();
            }
            queue[tail] = e;
            tail++;
            if (tail == queue.length) {tail = 0;}
             curr=size.getAndIncrement();

            //在方法内部
            if (curr+1<queue.length){
                tailsWait.signal();
            }

        } finally {
            tailLock.unlock();
        }
        //唤醒 待取值集合
        //插入值时 0 到 1 时 通知来取值 剩下的情况在poll方法内
        if (curr==0) {
            try {
                headLock.lock();
                headsWait.signal();
            } finally {
                headLock.unlock();
            }
        }

    }

    @Override
    public E poll() throws InterruptedException {
        //上头锁取值
        E e;
        int curr;
        headLock.lockInterruptibly();
        try {
            while (isEmpty()) {
                headsWait.await();
            }
             e = queue[head];
            queue[head] = null;
            head++;
            if (head == queue.length) {head = 0;}
            curr=size.getAndDecrement();

            //剩下的情况在poll方法内
            if (curr >1){
                headsWait.signal();
            }

        } finally {
            headLock.unlock();
        }

        //唤醒 待插入集合 尽可能减少上锁
        //从满队列取出值时 通知等待线程 插入 剩下的情况在offer 内部 插入


        //从满队列取出值时 通知 插入  其它的情况在 阻塞的插入线程在offer方法内判断是否要插入
        if (curr== queue.length) {
            try {
                tailLock.lock();
                tailsWait.signal();
            } finally {
                tailLock.unlock();
            }
        }


        return e;
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        return false;
    }




    public static void main(String[] args) {
        final TestBlockingQueue2<String> queue = new TestBlockingQueue2<>(5); // 创建一个容量为5的阻塞队列
        ExecutorService executor = Executors.newFixedThreadPool(4); // 创建一个线程池

        // 生产者
        Runnable producer = () -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String item = "Item " + i;
                    queue.offer(item);
                    System.out.println(Thread.currentThread().getName() + " produced: " + item);
                    Thread.sleep(500); // 模拟生产时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        // 消费者
        Runnable consumer = () -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String item = queue.poll();
                    System.out.println(Thread.currentThread().getName() + " consumed: " + item);
                    Thread.sleep(1000); // 模拟消费时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        // 启动多个生产者和消费者线程
        for (int i = 0; i < 2; i++) {
            executor.submit(producer);
            executor.submit(consumer);
        }

        executor.shutdown(); // 关闭线程池
    }
















}
